package org.NooLab.glue.recdispatcher;

import java.util.Vector;

 
import org.NooLab.glue.MessageBoard;
import org.NooLab.glue.interfaces.SynchronizedMessageDeliveryIntf;
import org.NooLab.glue.msgframe.MessageHandlerAbs;
import org.NooLab.glue.transaction.Transaction;
import org.NooLab.utilities.logging.PrintLog;




/**
 * 
 * The purpose of this class is to organize the dispatch a set of tasks to a set of receptors 
 * such that all receptors are addressed, without overlapping the incoming tasks (which are flowing in in parallel)
 * 
 * It is thus used only for TCP/UDP !! for http, the parallelity is automatically established through the independence of the clients 
 * 
 * in  SourceMessageInterpreter::sourceVariant_DataRelayToReceptor() items are added
 * 
 * This requires careful supervision of the sending task;
 * in fact, the allocation needs to be separated from the actual sending. 
 * 
 * it maintains a list for the assignment task->receptor and also contains 
 * the respective dispatcher (inner class SenderHandler)<br/>
 * 
 * only the receptor itself can remove itself from this list !!!
 *
 * this class is getting constructed y the MessageBoard, which guarantees its uniqueness
 * 
 */
public class ReceptorLinks{
	
	
	
	Vector<ReceptorLink> items = new Vector<ReceptorLink>();
	Vector<Integer> activatedPorts = new Vector<Integer>(); 
	
	
	SenderHandler senderHandler;
	
	MessageBoard mb;
	SynchronizedMessageDeliveryIntf smcb ;
	
	int localThreadsCount = 2;
	int nrOfProcessors = 1;
	
	int dispatcherPause =0;
	
	PrintLog out;
	
	public ReceptorLinks( MessageBoard mb, PrintLog outprn){
		
		
		this.mb = mb;
		out = outprn;

		Runtime runtime = Runtime.getRuntime();
        nrOfProcessors = runtime.availableProcessors();
        localThreadsCount = nrOfProcessors - 1 ;
        if (localThreadsCount<=0){ localThreadsCount=1;}
        
        										out.print(3, nrOfProcessors+" processors found, "+localThreadsCount+" receptor processes created.");
		senderHandler = new SenderHandler( mb,out );
	}
	
	public void setCallback( SynchronizedMessageDeliveryIntf cb ){
		smcb = cb;
	}
	
	
	public int size(){
		return items.size();
	}
	
	synchronized public void put( Transaction transaction, String xmlmsg, int deliveryPort){
		ReceptorLink item = new ReceptorLink();
		ReceptorLink rit;
		
		item.receptorPort = deliveryPort;
		item.xmlStr = xmlmsg ;
		item.transaction = transaction;
		
		for (int i=0;i<items.size();i++){
			rit = items.get(i) ;
			if ((rit.receptorPort==deliveryPort) && (rit.transaction.getTransactionID()==transaction.getTransactionID())){
				return;
			}
		}
		// avoid double issues
		items.add(item) ;
	}
	
	public Vector<Integer> getActivatedPorts() {
		return activatedPorts;
	}

	public void removeByTransactionID( long transactionID ){
		
		int p,ix;
		
		while (senderHandler.isWorking()){
			out.delay(1);
		}
		dispatcherPause=1;
		
		for (int i=0;i<items.size();i++){
		
			if ( transactionID == items.get(i).transaction.getTransactionID()){
				p = items.get(i).receptorPort ;
				ix = activatedPorts.indexOf(p) ;
				if (ix>=0){
					activatedPorts.remove(ix);
				}
				items.remove(i);
				break;
			}
		}
		
		dispatcherPause = 0;
	}
	
	
	public ReceptorLink get(int index){
		ReceptorLink reli = null;
		
		reli = items.get(index) ;
		
		return reli ;
	}
	
	public void remove(int index){
		ReceptorLink reli = null;
		
		
	}
	

	
	class SenderHandler extends MessageHandlerAbs {
		
		
		PrintLog out;
		
		// move as much as possible to the abstract class
		public SenderHandler( MessageBoard mb, PrintLog outprn ){
			super(mb,"SenderHandler");
			// there the thread also starts !
			
			out = outprn;
			out.print(3,"senderhandler has been started...");
			thrd.setPriority( Thread.NORM_PRIORITY + 2);
		}

		 
		
		public void run() {
			int d=0, portnumber; 
			isRunning = true;
			ReceptorLink item ;
			long tid;
			
			try{

				while ((isRunning) ) {
					
					if ((isWorking == false) && (dispatcherPause==0) && (out!=null) && (items!=null) ){
						isWorking = true;
										out.print(6,">>>   SenderHandler");
						if (items.size() > 0) {
							d=10;
										
								
							// work on it... on the whole list !!
							// item = items.get(0); // -> this would activate only the first, and since this get "blocked" due to usage
								 				    // (it is listed in activatedPorts) no other item will be selected
							// thus we have a loop here across all items
							
							int i=0;
							while (i<items.size()){
							 
								if (activatedPorts.size() < localThreadsCount){
									item = items.get(i); // clients
								}else{
									// only 1 client is served, and since only the client can remove itself from the list
									// of activated ports = receptorLinks, the clients will be activated one AFTER another !!
									item = items.get(0); 
								}
								portnumber = item.receptorPort ;
							
								if (( activatedPorts.indexOf(portnumber)<0) && (dispatcherPause==0)){
												
												out.printErr(3,	"from a set of "+items.size()+" items, "+
																"assigning item (tid:"+item.transaction.getTransactionID()+") "+
																"to port (p:"+portnumber+") ");
									// see the condition above: no double activation is possible any more
								    activatedPorts.add(portnumber) ;
								    
								    int tsv = item.transaction.getTransactState();
								    if (tsv<Transaction.TS_TASK_DIGESTING){
								    	
								    	// hosted in MessageBoard, because there should be only 1 instance of it
				     /*>>>>>>>>>>>>>>*/ smcb.synchedsend( item ) ; // over there, the thread is treated by adhoc instances of
								    	
								    	// a simple class in order to separate the data nicely & safely
								    	
								    }else{
								    	removeByTransactionID( item.transaction.getTransactionID());
								    	int ix = activatedPorts.indexOf(portnumber);
								    	if (ix>=0){ activatedPorts.remove(ix);}
								    	i--;
								    }
								      
								}
								// only after state signaling by the receptor the port will be removed 
								// again from this list, i.e. until then it remains blocked, while at the same
								// time all available receptors are found and assigned
								out.delay(2); // a very small delay between send attempts.
							
								i++;
							} // i-> all items
						}
						isWorking = false;
					}  
					out.delay(10) ;
					out.delay(1) ;
					
				} // while ...
				
			}catch(Exception e){
				e.printStackTrace(); 
			}finally{
				isRunning = false;
				isWorking = false;
			}
		}
			 
	} // SenderHandler 
}
